package com.qingcloud.df.executor.worker;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.qingcloud.base.constant.SystemConstant;
import com.qingcloud.base.remote.executor.ExecutorParam;
import com.qingcloud.df.executor.queue.publisher.ComponentPublisher;
import com.qingcloud.df.executor.utils.ComponentUtil;
import com.qingcloud.df.sdk.component.config.OutputDataItem;
import com.qingcloud.df.sdk.component.constains.Constants;
import com.qingcloud.df.sdk.component.context.CompInstContext;
import com.qingcloud.df.sdk.component.context.ConfigManager;
import com.qingcloud.df.sdk.component.context.LoggerContext;
import com.qingcloud.df.sdk.component.context.NodeManager;
import com.qingcloud.df.sdk.component.message.*;
import com.qingcloud.df.sdk.component.message.request.*;
import com.qingcloud.df.sdk.component.message.response.AkkaResponse;
import com.qingcloud.df.sdk.component.message.response.ExecutorStatus;
import com.qingcloud.df.sdk.component.message.response.JobResponse;
import com.qingcloud.df.sdk.component.transport.actor.*;
import com.qingcloud.df.sdk.component.transport.handler.ExecutorHandlerUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;

@Service
@AkkaActor(path = Constants.E4G_PATH)
public class ExecutorActorHandler {

    Logger logger = LoggerFactory.getLogger(ExecutorActorHandler.class);
    @Resource
    private ExecutorService executorService;
    /**
     * 执行器数据传输 需要下载更新配置文件 Server-->Executor  (任务分配唯一入口）
     * @param request
     */
    @AkkaHandler(path = Constants.SE_HANDLER_ASSIGIN_JOB)
    public void  assiginJob(JobInstanceRequest request)  {
        logger.debug("------------------->receive assiginJob data<--------------------");
        List<CompInstContext> instances = request.getCompInstList();
        if(instances!=null){
            executorService.destroyJob(request.getJobId(), request.getEnv());
            for(CompInstContext instance : instances){
                ExecutorParam.InstanceParam instanceParam = new ExecutorParam.InstanceParam();
                instanceParam.setInstanceId(instance.getId());
                instanceParam.setJobId(instance.getJobId());
                instanceParam.setEnv(request.getEnv());
                instanceParam.setCron(instance.getCron());
                String objectName = String.format(SystemConstant.JOB_CONFIG_OBJECTNAME,instance.getEnv(),instance.getId().toString(),instance.getSoftwareName());
                instanceParam.setConfigUrl(objectName);
                instanceParam.setTaskId(instance.getTaskId());
                //启动实例
                executorService.startInstance(instanceParam);
            }
            List<CompInstContext>  contexts = ConfigManager.loadByJobId(request.getJobId(),request.getEnv());
            NodeManager.put(request.getJobId(),contexts);
        }
        JobResponse response = new JobResponse();
        response.setJobId(request.getJobId());
        response.setContextMap(NodeManager.getInstanceByJobId(request.getJobId()));
        ExecutorHandlerUtil.sendJobContext(response);
    }

    /**
     * 新增或更新任务实例 需要下载更新配置文件
     * @param context
     * @return
     * @throws Exception
     */
    @AkkaHandler(path = Constants.SE_HANDLER_ASSIGIN_INSTANCE)
    public void assiginInstance(CompInstContext context) {
        logger.debug("------------------->receive assiginInstance data<--------------------");
        ExecutorParam.InstanceParam instanceParam = new ExecutorParam.InstanceParam();
        instanceParam.setJobId(context.getJobId());
        instanceParam.setInstanceId(context.getId());
        instanceParam.setEnv(context.getEnv());
        instanceParam.setCron(context.getCron());
        String objectName = String.format(SystemConstant.JOB_CONFIG_OBJECTNAME,context.getEnv(),context.getId().toString(),context.getSoftwareName());
        instanceParam.setConfigUrl(objectName);
        instanceParam.setTaskId(context.getTaskId());
        CompInstContext originContext = NodeManager.getContextByInstanceId(context.getId());
        //先移除现有的配置文件，停止服务
        if(originContext!=null){
            executorService.destroy(Arrays.asList(context.getId()),context.getEnv());
        }
        //再次启动实例
        executorService.startInstance(instanceParam);
        CompInstContext compInstContext = ConfigManager.loadByInstanceId(context.getId(),context.getEnv());
        NodeManager.add(context.getJobId(),compInstContext);

        JobResponse response = new JobResponse();
        response.setJobId(context.getJobId());
        response.setContextMap(NodeManager.getInstanceByJobId(context.getJobId()));
        ExecutorHandlerUtil.sendJobContext(response);
    }

    @AkkaHandler(path = Constants.SE_HANDLER_JOB_START)
    public AkkaResponse<JobResponse> jobStart(JobStartRequest request) {
        logger.debug("------------------->receive jobStart  data<--------------------");
        List<ExecutorStatus> executorStatuses = executorService.startJob(request.getJobId(),request.getEnv());
        return buildResult(executorStatuses,request.getJobId(),request.getEnv());
    }
    @AkkaHandler(path = Constants.SE_HANDLER_JOB_STOP)
    public AkkaResponse<JobResponse> jobStop(JobStopRequest request) {
        logger.debug("------------------->receive jobStop  data<--------------------");
        List<ExecutorStatus> executorStatuses = executorService.stopJob(request.getJobId(),request.getEnv());
        return buildResult(executorStatuses,request.getJobId(),request.getEnv());
    }

    @AkkaHandler(path = Constants.SE_HANDLER_JOB_DESTROY)
    public AkkaResponse<JobResponse> jobDestroy(JobRemoveRequest request) {
        logger.debug("------------------->receive jobDestroy  data<--------------------");
        List<ExecutorStatus> executorStatuses = executorService.destroyJob(request.getJobId(),request.getEnv());
        return buildResult(executorStatuses,request.getJobId(),request.getEnv());
    }

    @AkkaHandler(path = Constants.SE_HANDLER_INSTANCE_START)
    public AkkaResponse<JobResponse> instanceStart(InstanceStartRequest request) {
        logger.debug("------------------->start instanceStart process data<--------------------");
        ExecutorParam.InstanceParam instanceParam = new ExecutorParam.InstanceParam();
        instanceParam.setInstanceId(request.getInstanceId());
        instanceParam.setJobId(request.getJobId());
        instanceParam.setConfigUrl(request.getConfigPath());
        instanceParam.setEnv(request.getEnv());
        ExecutorStatus executorStatuses = executorService.startInstance(instanceParam);
        return buildInstResult(Arrays.asList(executorStatuses),null,request.getInstanceId(),request.getEnv());
    }

    @AkkaHandler(path = Constants.SE_HANDLER_INSTANCE_STOP)
    public AkkaResponse<JobResponse> instanceStop(InstanceStopRequest request) {
        logger.debug("------------------->stop instanceStop process data<--------------------");
        List<ExecutorStatus> executorStatuses = executorService.stopInstance(request.getInstanceId(),request.getEnv());
        return buildInstResult(executorStatuses,null,request.getInstanceId(),request.getEnv());
    }

    @AkkaHandler(path = Constants.SE_HANDLER_INSTANCE_DESTROY)
    public AkkaResponse<JobResponse> instanceRemove(InstanceRemoveRequest request) {
        logger.debug("------------------->stop instanceRemove process data<--------------------");
        CompInstContext context = NodeManager.getContextByInstanceId(request.getInstanceId());
        Integer jobId = context.getJobId();
        List<ExecutorStatus> executorStatuses = executorService.destroy(Collections.singletonList(request.getInstanceId()),request.getEnv());

        NodeManager.removeInst(request.getInstanceId());
        return buildInstResult(executorStatuses,jobId,request.getInstanceId(),request.getEnv());
    }

    private AkkaResponse<JobResponse> buildResult(List<ExecutorStatus> executorStatuses,Integer jobId,String env){
        List<CompInstContext>  contexts = ConfigManager.loadByJobId(jobId, env);
        NodeManager.put(jobId,contexts);
        JobResponse response= new JobResponse();
        response.setJobId(jobId);
        response.setContextMap(NodeManager.getInstanceByJobId(jobId));
        response.setStatusList(executorStatuses);
        return AkkaResponse.success(response);
    }

    private AkkaResponse<JobResponse> buildInstResult(List<ExecutorStatus> executorStatuses,Integer jobId, Integer instanceId,String env){
        CompInstContext compInstContext = ConfigManager.loadByInstanceId(instanceId,env);
        if(compInstContext!=null && compInstContext.getJobId()!=null) {
            NodeManager.add(compInstContext.getJobId(), compInstContext);
            jobId = compInstContext.getJobId();
        }
        JobResponse response= new JobResponse();
        response.setJobId(jobId);
        response.setContextMap(NodeManager.getInstanceByJobId(jobId));
        response.setStatusList(executorStatuses);
        return AkkaResponse.success(response);
    }

    @AkkaHandler(path = Constants.ES_HANDLER_SERVER_PING)
    public AkkaResponse<Pong>  pong(Ping ping) {
        logger.debug("------------------->receive server ping request<--------------------");
        Pong pong = NodeService.serverPing(ping);
        return AkkaResponse.success(pong);
    }


    /**
     * 执行器数据传输 Executor-->Executor
     * @param reportLog
     */
    @AkkaHandler(path = Constants.EE_HANDLER_PROCESS)
    public void executorProcess(ReportLog reportLog) {
        logger.debug("------------------->receive executor process data<--------------------");
    }

    /**
     * 数据接收器网关-日志汇报  Gateway--> Executor
     * @param reportLogs
     */
    @AkkaHandler(path = Constants.GE_HANDLER_REPORTLOG)
    public void gatewayReportLog(ReportLogs reportLogs) {
        logger.debug("------------------->receive gateway report logs<--------------------");
        if(reportLogs != null ){
            Integer instanceId = reportLogs.getInstanceId();
            List<ReportLog> reportLogList = reportLogs.getReportLogList();
            reportLogList.forEach(reportLog -> {
                LoggerContext.sender(instanceId,reportLog.getMessage());
            });
        }
    }

    /**
     * 数据接收器网关-数据上报  Gateway--> Executor
     * @param reportData
     */
    @AkkaHandler(path = Constants.GE_HANDLER_REPORTDATA)
    public void gatewayReportData(ReportData reportData) {
        logger.debug("------------------->receive gateway report data<--------------------");

        if(reportData != null ){
            Integer instanceId = reportData.getInstanceId();
            String env  =  reportData.getEnv();
            String messageFrom  =  reportData.getAkkaAddress();
            List<ComponentValue> reportDataList = reportData.getReportDataList();
            CompInstContext context = ConfigManager.loadByInstanceId(instanceId,env);
            LoggerContext.sender(instanceId,"----》【执行中心】接收到来自网关数据成功-"+instanceId+"《-----------------");
            if(context==null){
                logger.error("can not find this component instance system config 【{}】-【{}】!",instanceId,env);
                return ;
            }
            if(reportDataList!=null){
                reportDataList.forEach(componentValue->{
                    LoggerContext.sender(instanceId,"----》【执行中心】接收到来自网关本批次数据:《-----------------");
                    List<Map<String, OutputDataItem>> values = componentValue.getValues();
                    String jsonFormatString = JSON.toJSONString(values, SerializerFeature.PrettyFormat,
                            SerializerFeature.WriteMapNullValue,
                            SerializerFeature.WriteDateUseDateFormat);
                    LoggerContext.sender(instanceId,jsonFormatString);
                    LoggerContext.sender(instanceId,"----》【执行中心】接收到来自网关本批次数据结束:《-----------------");
                    ComponentPublisher componentPublisher = ComponentUtil.getComponentPublisher(context);
                    componentPublisher.send(context.getTopic(),messageFrom,JSON.toJSONString(componentValue));
                });
            }
            LoggerContext.sender(instanceId,"----》【执行中心】数据发送成功-"+instanceId+"《-----------------");
        }
    }

    /**
     * 数据接收器网关-健康状况汇报   Gateway-->Executor
     * @param reportHealth
     */
    @AkkaHandler(path = Constants.GE_HANDLER_HEARTBEAT)
    public void gatewayReportHealth(ReportHealth reportHealth) {
        logger.debug("------------------->receive gateway report healthy<--------------------");
    }
}
